package com.abyss.checkpoint;


import org.apache.flink.api.common.restartstrategy.RestartStrategies;
import org.apache.flink.api.common.time.Time;
import org.apache.flink.runtime.state.filesystem.FsStateBackend;
import org.apache.flink.streaming.api.CheckpointingMode;
import org.apache.flink.streaming.api.environment.CheckpointConfig;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

/**
 * 检查点的一些设置
 */
public class AllCheckPointConfigDemo {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        // 开启检查点 参数1是检查点的执行间隔 参数2是执行的一致性是啥
        env.enableCheckpointing(5000L, CheckpointingMode.EXACTLY_ONCE);
        // 设置检查点的保存后端，我们经常用的是FsStateBackend保存到文件系统可以支持本地文件和hdfs文件
        env.setStateBackend(new FsStateBackend("file:///data/output/checkpoint"));
        // 设置重启策略为：固定延迟重启策略，参数1：整个job允许几次失败，参数2：失败后等待多久重启task
        env.setRestartStrategy(RestartStrategies.fixedDelayRestart(5, 5000L));
        // 设置重启策略为：失败率策略，参数1：单位时间内最大失败次数，参数2：单位时间，参数3：失败后等多久重启下一个任务
        // 这个就是表示，1分钟内失败3次整个JOB停止。每次失败后等待5秒重启失败的task
        env.setRestartStrategy(RestartStrategies.failureRateRestart(3, Time.minutes(1), Time.seconds(5)));
        // 重启策略为：无策略，当出现Task失败，整个JOB 停止
        env.setRestartStrategy(RestartStrategies.noRestart());
        // 拿到检查点的配置文件对象
        CheckpointConfig checkpointConfig = env.getCheckpointConfig();
        // 设置检查点在任务取消的时候是否保留，默认是删除不保留，如果需要保留，按照如下设置
        checkpointConfig.enableExternalizedCheckpoints(CheckpointConfig.ExternalizedCheckpointCleanup.RETAIN_ON_CANCELLATION);
        // 设置同一时间下，并行执行的检查点任务的最大数量
        checkpointConfig.setMaxConcurrentCheckpoints(1);
        // 检查点超时，单位毫秒
        checkpointConfig.setCheckpointTimeout(5000L);
        // 两个检查点之间 必须间隔的时间
        checkpointConfig.setMinPauseBetweenCheckpoints(1000L);
        // 设置检查点的执行间隔，在env.enableCheckpointing里面已经设置过了
        checkpointConfig.setCheckpointInterval(5000L);
        // 设置检查点的模式, 在env.enableCheckpointing里面已经设置过了
        checkpointConfig.setCheckpointingMode(CheckpointingMode.EXACTLY_ONCE);

    }
}